package org.redis.cluster.redis.transport;

import org.apache.commons.lang3.ArrayUtils;
import org.redis.cluster.RedisCluster;
import org.redis.cluster.extension.ExtensionLoader;
import org.redis.cluster.interceptor.Invocation;
import org.redis.cluster.interceptor.InvocationFactory;
import org.redis.cluster.plugin.Plugin;
import org.redis.cluster.redis.RedisCommand;
import org.redis.cluster.redis.processor.RedisReplyProcessorHelper;
import org.redis.cluster.redis.transport.reply.RedisErrorReply;
import org.redis.cluster.transport.event.SessionClosedEvent;
import org.redis.cluster.transport.event.SessionCreatedEvent;
import org.redis.cluster.transport.session.Session;
import org.redis.cluster.transport.session.SessionProcessor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class RedisServerSessionProcessor implements Plugin,SessionProcessor<RedisCommand> {
	
	private final static String INVOCATION_FACTORY="invocationFactory";
	
	private final static String PROCESSOR="serverProcessor";
	
	protected final Logger logger=LoggerFactory.getLogger(getClass());
	
	private RedisCluster redisCluster;
	private ExtensionLoader extensionLoader;
	private InvocationFactory invocationFactory;
	private RedisServerProcessor processor;
	
	private RedisReplyProcessorHelper replyProcessor;
	
	@Override
	public void init(RedisCluster redisCluster) {
		this.redisCluster=redisCluster;
		this.extensionLoader=redisCluster.getExtensionLoader();
		this.invocationFactory=initInvocationFactory();
		this.processor=initProcessor();
		this.replyProcessor=initRedisReplyProcessor();
	}
	
	protected InvocationFactory initInvocationFactory() {
		return extensionLoader.load(INVOCATION_FACTORY,InvocationFactory.class);
	}
	
	protected RedisServerProcessor initProcessor() {
		return extensionLoader.load(PROCESSOR,RedisServerProcessor.class);
	}
	
	protected RedisReplyProcessorHelper initRedisReplyProcessor() {
		return new RedisReplyProcessorHelper(redisCluster);
	}
	
	@Override
	public void destroy() {
	}

	@Override
	public void process(Session session, RedisCommand command) {
		Invocation invocation=invocationFactory.getInvocation(session, command, ()->{
			return processor.process(session, command);
		});
		
		RedisReply reply=null;
		
		try {
			reply=invocation.invoke();
		}
		catch(Throwable t) {
			if(logger.isErrorEnabled()) {
				logger.error("",t);
			}
		}
		
		if(reply==null) {
			return;
		}
		
		byte replyByteArray[]=replyProcessor.process(reply);
		if(ArrayUtils.isNotEmpty(replyByteArray)) {
			session.writeAndFlush(replyByteArray);
		}
	}

	@Override
	public void created(Session session) {
		redisCluster.getEventPusher().push(new SessionCreatedEvent(session));
	}

	@Override
	public void closed(Session session) {
		redisCluster.getEventPusher().push(new SessionClosedEvent(session));
	}

	@Override
	public void exception(Session session, Throwable cause) {
		if(logger.isErrorEnabled()) {
			logger.error("",cause);
		}
		
		session.write(new RedisErrorReply(cause.getMessage()));
	}
}